SpringBoot整合gRPC工程搭建,实现远程服务调用 |
您所在的位置：网站首页 › grpc 调用 › SpringBoot整合gRPC工程搭建,实现远程服务调用 |
SpringBoot整合gRPC工程搭建,实现远程服务调用
|
本工程基于Springboot 2.5.4，gRPC的1.40.0版本搭建。 gRPC使用Http2协议，具体实现grpc provider端必须依赖grpc-netty或grpc-netty-shaded来提供服务， grpc消费端可以通过grpc-netty调用,也可以通过grpc-okhttp调用。 如果一个应用既是provider又是consumer，建议只需要依赖grpc-netty即可，本文采用netty来作为gRPC的底层服务实现. 细心的话会发现这个工程搭建没有用到注册中心，原因是gRPC没有实现对接注册中心，需要自己通过gRPC的扩展点对接中心，本节暂不做讲解 工程主要分为四个部分 grpc_consumer：服务消费端 grpc_provider：服务提供端 proto_api：公共api，主要存放grpc需要的.proto文件，并动态生成java代码 pom.xml：parent pom文件，统一管理依赖包版本，见本文最后
在src/mian目录下创建proto目录用于存在.proto文件，在proto目录下创建RpcService.proto文件用于定义gRPC服务，为了方便编写proto文件，可以在idea中安装 Protobuf 插件，可以根据.proto文件生成gRPC支持的任意语言的代码 syntax = "proto3"; option java_multiple_files = true; //生成的代码不要在一个文件中,生成多个java文件 option java_package = "com.bruce.grpc.proto.service"; option java_outer_classname = "RpcServiceProto"; //默认等于proto文件名的驼峰命名 //服务请求实体 message HelloRequest { string greeting = 1; repeated bytes data = 2; string serial_type = 3; } //服务返回值实体 message HelloResponse { string reply = 1; repeated bytes data = 2; string serial_type = 3; } //定义服务 service HelloService { //定义服务方法, 动态生成的java方法会自动改成驼峰命名方式 rpc SayHello(HelloRequest) returns (HelloResponse); } 配置maven插件，根据.proto文件动态生成代码配置maven插件，根据.proto文件动态生成grpc服务代码，grpc代码生成后编译不报错最少依赖grpc-protobuf,grpc-stub，但是如果要启动服务，在provider端还需要依赖netty和grpc-netty。消费端消费服务时需要依赖netty和grpc-netty或者只是做消费调用的话单独依赖grpc-okhttp也可以。 com.bruce.grpc springboot_grpc 0.0.1 jar proto_api 0.0.1-SNAPSHOT proto_api proto-api 1.8 io.grpc grpc-protobuf io.grpc grpc-stub kr.motd.maven os-maven-plugin org.xolstice.maven.plugins protobuf-maven-plugin com.google.protobuf:protoc:3.17.2:exe:${os.detected.classifier} grpc-java io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier} compile compile-custom执行compile命令之后,会在target目录生成grpc实体类和服务代码，服务代码只是一个抽象实现，需要业务自己具体的业务逻辑，(在grpc_provider模块实现）。
用于Spring扫描gRPC服务实现，注册成Spring的Bean @Target(ElementType.TYPE) @Retention(RetentionPolicy.RUNTIME) @Documented @Component public @interface GrpcService { @AliasFor(annotation = Component.class) String value() default ""; } 实现rpc服务HelloServiceGrpc.HelloServiceImplBase是动态生成的抽象类，里面生成了我们在.proto文件中定义的服务方法，实现该方法即可。 使用@GrpcService让Spring扫描到服务实现，等gRPC Server端启动之前从Spring中获取所有的gRPC服务实现，注册到gRPC Server等待客户端调用。 import com.bruce.grpc.proto.service.HelloRequest; import com.bruce.grpc.proto.service.HelloResponse; import com.bruce.grpc.proto.service.HelloServiceGrpc; import com.bruce.provider.annotation.GrpcService; import io.grpc.stub.StreamObserver; import lombok.extern.slf4j.Slf4j; /** * Created by bruce on 2021/8/26 22:08 */ @GrpcService @Slf4j public class HelloServiceImpl extends HelloServiceGrpc.HelloServiceImplBase { @Override public void sayHello(HelloRequest request, StreamObserver responseObserver) { //不能调用父类，否则报错，目的就是告诉你一定要自己实现业务逻辑 //super.sayHello(request, responseObserver); String greeting = request.getGreeting(); log.info(greeting); HelloResponse response = HelloResponse.newBuilder() .setReply("hello你好啊") .build(); responseObserver.onNext(response); //responseObserver.onNext(response); responseObserver.onCompleted(); } } 实现gRPC Server端注册RPC服务具体实现，启动服务，等待远程调用 /** * Created by bruce on 2021/7/15 19:53 */ @Slf4j public class GRpcServer { private Server server; public void start(List services) { // MyGrpcServerInterceptor myGrpcServerInterceptor = new MyGrpcServerInterceptor(); MutableHandlerRegistry mutableHandlerRegistry = new MutableHandlerRegistry(); for (BindableService service : services) { // 用于添加拦截器 // ServerServiceDefinition interceptedService = ServerInterceptors.intercept(service, myGrpcServerInterceptor); mutableHandlerRegistry.addService(service); log.info("Add grpc service impl [{}]", service.getClass()); } try { server = ServerBuilder.forPort(50051) .fallbackHandlerRegistry(mutableHandlerRegistry) //.addService(new HelloServiceImpl()) .build() .start(); } catch (IOException e) { e.printStackTrace(); } log.info("GRpc Server started, listening on " + 50051); Runtime.getRuntime().addShutdownHook(new Thread(() -> { // Use stderr here since the logger may have been reset by its JVM shutdown hook. System.out.println("*** shutting down gRPC server since JVM is shutting down"); try { GRpcServer.this.stop(); } catch (InterruptedException e) { e.printStackTrace(System.err); } System.out.println("*** server shut down"); })); } public Server getServer() { return server; } public void awaitTermination() { try { server.awaitTermination(); } catch (InterruptedException e) { e.printStackTrace(); } } public void stop() throws InterruptedException { if (server != null) { server.shutdown().awaitTermination(30, TimeUnit.SECONDS); } } } 启动gRPC Server端 import com.bruce.provider.annotation.GrpcService; import io.grpc.BindableService; import org.springframework.beans.BeansException; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; import java.util.ArrayList; import java.util.Map; /** * Created by bruce on 2021/8/26 00:18 */ @Component public class ComponentBean implements ApplicationContextAware, SmartInitializingSingleton, DisposableBean { private ApplicationContext applicationContext; @Bean public GRpcServer gRpcServer() { return new GRpcServer(); } @Override public void setApplicationContext(ApplicationContext context) throws BeansException { this.applicationContext = context; } @Override public void afterSingletonsInstantiated() { Map grpcServices = applicationContext.getBeansWithAnnotation(GrpcService.class); ArrayList services = new ArrayList(); for (Map.Entry entry : grpcServices.entrySet()) { if (!(entry.getValue() instanceof BindableService)) { throw new IllegalStateException("The bean named " + entry.getKey() + " is marked with the @GrpcService , must implement the " + BindableService.class.getName()); } services.add((BindableService) entry.getValue()); } GRpcServer gRpcServer = applicationContext.getBean(GRpcServer.class); gRpcServer.start(services); } @Override public void destroy() throws Exception { GRpcServer gRpcServer = applicationContext.getBean(GRpcServer.class); gRpcServer.stop(); } } 防止进程退出但是启动之后在没有其它非守护进程的情况下,进程便会退出。所以需要调用gRPC的io.grpc.Server#awaitTermination()来阻止进程退出. 但是该方法会阻塞当前线程,启动阶段也就是main线程，所以需要在SpringBoot代码启动完成后调用. @SpringBootApplication public class GrpcProviderApplication { public static void main(String[] args) { ConfigurableApplicationContext context = SpringApplication.run(GrpcProviderApplication.class, args); GRpcServer gRpcServer = context.getBean(GRpcServer.class); gRpcServer.awaitTermination(); } } grpc_consumer 服务消费端实现
如下便是一个最简单的客户端调用代码，通过io.grpc.stub.AbstractStub来调用远程服务，gRPC里面有三个抽象子类io.grpc.stub.AbstractFutureStub、io.grpc.stub.AbstractBlockingStub、io.grpc.stub.AbstractAsyncStub，示例中采用同步调用的方式，由插件生成的代码便是HelloServiceBlockingStub，但是该对象的创建需要一个客户端channel ManagedChannel作为参数。 import com.bruce.grpc.proto.service.HelloRequest; import com.bruce.grpc.proto.service.HelloResponse; import com.bruce.grpc.proto.service.HelloServiceGrpc; import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; import lombok.extern.slf4j.Slf4j; import java.io.IOException; import java.util.concurrent.TimeUnit; /** * Created by bruce on 2021/7/15 19:53 */ @Slf4j public class GRpcClient { public void start() throws IOException { ManagedChannelBuilder channelBuilder = ManagedChannelBuilder.forTarget("localhost:50051"); channelBuilder.usePlaintext(); //channelBuilder.defaultLoadBalancingPolicy("random") //设置 //channelBuilder.intercept(new MyClientInterceptor()); //CompressorRegistry compressorRegistry = CompressorRegistry.getDefaultInstance(); //compressorRegistry.register(null); //channelBuilder.compressorRegistry(compressorRegistry); //channelBuilder.decompressorRegistry() //channelBuilder.disableRetry(); //channelBuilder.idleTimeout(10,TimeUnit.MINUTES); //channelBuilder.keepAliveTimeout() ManagedChannel channel = channelBuilder.build(); HelloRequest request = HelloRequest.newBuilder() .setGreeting("this is a client request") .build(); HelloServiceGrpc.HelloServiceBlockingStub helloServiceBlockingStub = HelloServiceGrpc.newBlockingStub(channel); HelloResponse helloResponse = helloServiceBlockingStub.sayHello(request); System.out.println(helloResponse.getReply()); try { channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); } catch (InterruptedException e) { e.printStackTrace(); } } public static void main(String[] args) throws IOException { GRpcClient gRpcClient = new GRpcClient(); gRpcClient.start(); } } Consumer端和Spring整合如上的客户端调用较为加单,但是为了需要和Spring整合使用，需要将ManagedChannel配置成一个Bean以便于创建xxxStub时共用，因为xxxStub是线程安全的，所以也可以配置成Bean，在需要的地方通过@Autowired依赖注入使用。 /** * Created by bruce on 2021/8/26 00:32 */ @Configuration public class GRpcClientConfiguration { // 创建grpc客户端channel @Bean public ManagedChannel managedChannel() { ManagedChannel channel = ManagedChannelBuilder.forTarget("localhost:50051") .usePlaintext() .build(); return channel; } @Bean public HelloServiceGrpc.HelloServiceBlockingStub helloServiceBlockingStub(ManagedChannel channel) { return HelloServiceGrpc.newBlockingStub(channel); } } 消费端调用服务 @RestController public class HelloController { @Autowired private HelloServiceGrpc.HelloServiceBlockingStub helloServiceBlockingStub; @RequestMapping("/hello") public String hello() { HelloRequest request = HelloRequest.newBuilder().setGreeting("哈哈").build(); HelloResponse helloResponse = helloServiceBlockingStub.sayHello(request); String reply = helloResponse.getReply(); return reply; } } parent pom.xml org.springframework.boot spring-boot-starter-parent 2.5.4 com.bruce.grpc springboot_grpc 0.0.1 springboot_grpc pom Demo project for SpringBoot-gRPC 1.8 1.40.0 grpc_consumer grpc_provider proto_api io.grpc grpc-bom ${grpc.version} pom import io.grpc grpc-netty ${grpc.version} io.netty * kr.motd.maven os-maven-plugin 1.7.0 com.bruce.grpc proto_api 0.0.1-SNAPSHOT org.xolstice.maven.plugins protobuf-maven-plugin 0.6.1 |
【本文地址】
公司简介
联系我们
今日新闻 |
点击排行 |
|
推荐新闻 |
图片新闻 |
|
专题文章 |